/********************************************************/
/*	Copyright (C) 2012 Gong Li Bin			*/
/*	Project:	MapReduce			*/
/*	Author:		gong_libin			*/
/*	Date:		2013_05_06			*/
/*	File:		MrSquid.cpp			*/
/********************************************************/

#include "MrSquid.h"

bool g_bDebug = false;
MRSQUIDCONF_S g_stConf;

int main(int argc, char* argv[])
{
	CMrTcp CTcp;
	int iIndex = 0;
	int iReturn = 0;
	UCHAR ucType = MR_SD_NULL;
	char szConf[MR_BUFFER] = "/usr/local/bin/MrSquid.conf";

	struct option stOptions[] = {
		{"conf", 1, 0, 'c'},
		{"mysql", 0, 0, 'm'},
		{"pgsql", 0, 0, 'p'},
		{"oracle", 0, 0, 'o'},
		{"debug", 0, 0, 'd'},
		{"help", 0, 0, 'h'},
		{"version", 0, 0, 'v'},
		{0, 0, 0, 0},
	};

	while (true) {
		iReturn = getopt_long(argc, argv, "c:mpodhv", stOptions, &iIndex);
		if (iReturn < 0) {
			break;
		}
		switch (iReturn) {
			case 'c':	strncpy(szConf, optarg, MR_BUFFER);
					break;
			case 'm':	ucType = MR_SD_MSQL;
					break;
			case 'p':	ucType = MR_SD_PSQL;
					break;
			case 'o':	ucType = MR_SD_OSQL;
					break;
			case 'd':	g_bDebug = true;
					break;
			case 'h':	MrSquidHelp(argv[0]);
					return MR_SUCCESS;
			case 'v':	MR_PRINT("Copyright (C) 2012 Gong Li Bin	\n");
					MR_PRINT("%s Build on %s %s.\n", MR_SD_VERSION, __DATE__, __TIME__);
					return MR_SUCCESS;
			default:	return MR_SUCCESS;
		}
	}

	if (MR_SD_NULL == ucType) {
		MrSquidHelp(argv[0]);
		return MR_SUCCESS;
	}

	signal(SIGSEGV, MrSquidQuit);
	signal(SIGTERM, MrSquidQuit);

	if (true == g_bDebug) {
		MR_DEBUG("Initialize config from %s\n", szConf);
	}
	if (MR_SUCCESS == MrSquidConfig(ucType, szConf)) {
		if (true == g_bDebug) {
			MR_DEBUG("Type: %d\n", ucType);
			MR_DEBUG("Host: %s\n", g_stConf.m_szHost);
			MR_DEBUG("Port: %s\n", g_stConf.m_szPort);
			MR_DEBUG("User: %s\n", g_stConf.m_szUser);
			MR_DEBUG("Pass: %s\n", g_stConf.m_szPass);
			MR_DEBUG("Listen: %s\n", g_stConf.m_szListen);
		}
	}
	else {
		MR_ERROR("Failed to config\n");
		return MR_FAILURE;
	}

	if (true != g_bDebug) {
		MrFork();
	}

	if (true == g_bDebug) {
		MR_DEBUG("Initialize tcp listen %s...\n", g_stConf.m_szListen);
	}
	if (MR_SUCCESS == CTcp.MrTcpInit(g_stConf.m_szListen)) {
		CTcp.MrTcpSetFunction(MrSquidFunction);
		CTcp.MrTcpAccept();
		CTcp.MrNetClose();
	}

	return MR_SUCCESS;
}

void MrSquidHelp(char* pszProc)
{
	MR_PRINT("Usage: %s [options]\n", pszProc);
	MR_PRINT("Valid options are:\n");
	MR_PRINT("[*] -c or --conf:       set configure.\n");
	MR_PRINT("[0] -m or --mysql:      set mysql type.\n");
	MR_PRINT("[0] -p or --pgsql:      set pgsql type.\n");
	MR_PRINT("[0] -o or --oracl:      set oracl type.\n");
	MR_PRINT("[0] -d or --debug:      set debug mode.\n");
	MR_PRINT("[0] -h or --help:       show help.\n");
	MR_PRINT("[0] -v or --version:    show version.\n");
	MR_PRINT("Copyright (C) 2012 Gong Li Bin	\n");
	MR_PRINT("%s Build on %s %s.\n", MR_SD_VERSION, __DATE__, __TIME__);

	return;
}

void MrSquidQuit(int iSignal)
{
	if (true == g_bDebug) {
		MR_DEBUG("MrSquid is stopped by signal[%d]\n", iSignal);
	}

	exit(MR_FAILURE);
}

int MrSquidConfig(UCHAR ucType, char* pszConf)
{
	CMrConf CConf;
	char* pszType[] = {(char*)"", (char*)"MYSQL", (char*)"PGSQL", (char*)"ORACLE"};

	if (MR_SUCCESS == CConf.MrConfOpen(pszConf)) {
		if (MR_SUCCESS != CConf.MrConfGetVal(pszType[ucType], (char*)"HOST", g_stConf.m_szHost)) {
			goto MrSquidError;
		}
		if (MR_SUCCESS != CConf.MrConfGetVal(pszType[ucType], (char*)"PORT", g_stConf.m_szPort)) {
			goto MrSquidError;
		}
		if (MR_SUCCESS != CConf.MrConfGetVal(pszType[ucType], (char*)"USER", g_stConf.m_szUser)) {
			goto MrSquidError;
		}
		if (MR_SUCCESS != CConf.MrConfGetVal(pszType[ucType], (char*)"PASS", g_stConf.m_szPass)) {
			goto MrSquidError;
		}
		if (MR_SUCCESS != CConf.MrConfGetVal((char*)"GLOBAL", (char*)"LISTEN", g_stConf.m_szListen)) {
			goto MrSquidError;
		}
		CConf.MrConfClose();
	}
	else {
		return MR_FAILURE;
	}

	return MR_SUCCESS;

MrSquidError:
	CConf.MrConfClose();

	return MR_FAILURE;
}

void MrSquidFunction(int iSocket)
{
	CMrThread CThread;

	CThread.MrThreadAttrInit();
	CThread.MrThreadAttrSetDetach();
	CThread.MrThreadSetStartRoutine(MrSquidThreadCore);
	CThread.MrThreadCreate((void*)iSocket);

	return;
}

void* MrSquidThreadCore(void* pPara)
{
	CMrPro CPro;
	CMrTcp CTcp;
	int iLength = 0;
	CMrMysql CMysql;
	CMrPgsql CPgsql;
	CMrOracl COracl;
	CMrSql* pCSql = NULL;
	UCHAR uszPacket[MR_ETHERNET - MR_MACIPTCP] = { 0 };

	if (true == g_bDebug) {
		MR_DEBUG("Create thread: %lu\n", pthread_self());
	}

	CTcp.MrNetSetSocket((long)pPara);
	while ((iLength = CTcp.MrNetRecv(uszPacket, MR_ETHERNET - MR_MACIPTCP)) > 0) {
		CPro.MrProParse(uszPacket, iLength);
		if (true == g_bDebug) {
			MR_DEBUG("Receive packet: %d\n", iLength);
			CPro.MrProDisplay();
		}
		switch (CPro.MrProGetType()) {
			case MR_DEF_MYSL:
				pCSql = &CMysql;
				if (true == g_bDebug) {
					MR_DEBUG("Mysql: ");
				}
				break;
			case MR_DEF_PGSL:
				pCSql = &CPgsql;
				if (true == g_bDebug) {
					MR_DEBUG("Pgsql: ");
				}
				break;
			case MR_DEF_ORAL:
				pCSql = &COracl;
				if (true == g_bDebug) {
					MR_DEBUG("Oracle: ");
				}
				break;
			default:
				pCSql = NULL;
				if (true == g_bDebug) {
					MR_DEBUG("Error type\n");
				}
				break;
		}

		if (NULL != pCSql) {
			if (true == g_bDebug) {
				MR_DEBUG("Connect database...\n");
			}
			if (MR_SUCCESS == pCSql->MrSqlConnect(g_stConf.m_szHost, g_stConf.m_szPort, g_stConf.m_szUser, g_stConf.m_szPass)) {
				MrSquidSqlCore(pCSql, CTcp, CPro);
				if (true == g_bDebug) {
					MR_DEBUG("Close database...\n");
				}
				pCSql->MrSqlClose();
			}
		}
		else {
			MrSquidResultSend(CTcp, CPro, MR_DEF_UNKN, NULL, MR_PRO_0);
		}

		memset(uszPacket, '\0', MR_ETHERNET - MR_MACIPTCP);
	}
	CTcp.MrNetClose();

	if (true == g_bDebug) {
		MR_DEBUG("Destroy thread: %lu\n", pthread_self());
	}

	pthread_exit(NULL);
}

void MrSquidSqlCore(CMrSql* pCSql, CMrTcp& CTcp, CMrPro& CPro)
{
	int iLength = 0;
	USHORT usCount = 0;
	MRPROKV_S* pstKv = NULL;
	char szSql[MR_PACKET] = { 0 };

	if (true == g_bDebug) {
		MR_DEBUG("Sql total: %d\n", CPro.MrProGetTotal());
	}

	if (CPro.MrProGetTotal() > 0) {
		if (NULL != (pstKv = (MRPROKV_S*)MrMalloc(sizeof(MRPROKV_S) * CPro.MrProGetTotal()))) {
			MrProKvInput(pstKv, CPro.MrProGetTotal(), CPro.MrProGetBody(), CPro.MrProGetLength());
			while (usCount < CPro.MrProGetTotal()) {
				iLength = ((pstKv + usCount)->m_iLength < MR_PACKET) ? (pstKv + usCount)->m_iLength : MR_PACKET;
				strncpy(szSql, (char*)(pstKv + usCount)->m_puszValue, iLength);
				if (true == g_bDebug) {
					MR_DEBUG("Sql: %s\n", szSql);
				}
				switch (CPro.MrProGetType()) {
					case MR_DEF_MYSL:
						MrSquidMysqlCore(pCSql, CTcp, CPro, szSql);
						break;
					case MR_DEF_PGSL:
						MrSquidPgsqlCore(pCSql, CTcp, CPro, szSql);
						break;
					case MR_DEF_ORAL:
						MrSquidOraclCore(pCSql, CTcp, CPro, szSql);
						break;
					default:
						break;
				}
				memset(szSql, '\0', MR_PACKET);
				usCount += 1;
			}
			MrFree((void**)&pstKv);
		}
	}

	return;
}

void MrSquidMysqlCore(CMrSql* pCSql, CMrTcp& CTcp, CMrPro& CPro, char* pszSql)
{
	MYSQL_ROW stRow;
	USHORT usField = 0;
	void* pReturn = NULL;
	MRPROKV_S* pstKv = NULL;
	UCHAR ucReturn = MR_DEF_SUCC;

	if (true == g_bDebug) {
		MR_DEBUG("Verb: %d\n", CPro.MrProGetVerb());
	}

	switch (CPro.MrProGetVerb()) {
		case MR_DEF_SELT:
			pReturn = pCSql->MrSqlSelect(pszSql);
			break;
		case MR_DEF_INST:
			if (MR_SUCCESS != pCSql->MrSqlInsert(pszSql)) {
				ucReturn = MR_DEF_UNKN;
			}
			break;
		case MR_DEF_DELT:
			if (MR_SUCCESS != pCSql->MrSqlDelete(pszSql)) {
				ucReturn = MR_DEF_UNKN;
			}
			break;
		case MR_DEF_UPDT:
			if (MR_SUCCESS != pCSql->MrSqlUpdate(pszSql)) {
				ucReturn = MR_DEF_UNKN;
			}
			break;
		default:
			break;
	}

	if (MR_DEF_SELT == CPro.MrProGetVerb()) {
		if (NULL != pReturn && mysql_num_rows((MYSQL_RES*)pReturn) > 0) {
			if (NULL != (pstKv = (MRPROKV_S*)MrProKvMalloc(mysql_num_fields((MYSQL_RES*)pReturn), MR_PACKET))) {
				usField = 0;
				while (NULL != (stRow = mysql_fetch_row((MYSQL_RES*)pReturn))) {
					for (usField = 0; usField < mysql_num_fields((MYSQL_RES*)pReturn); usField ++) {
						(pstKv + usField)->m_ucKey = usField + 1;
						(pstKv + usField)->m_iLength = strlen((char*)stRow[usField]);
						memcpy((pstKv + usField)->m_puszValue, stRow[usField], strlen((char*)stRow[usField]));
					}
					MrSquidResultSend(CTcp, CPro, ucReturn, pstKv, usField);
					if (true == g_bDebug) {
						MR_PRINT("KeyValue:\n");
						MrProKvDisplay(pstKv, usField);
					}
					MrProKvReset(pstKv, mysql_num_fields((MYSQL_RES*)pReturn), MR_PACKET);
				}
				MrSquidResultSend(CTcp, CPro, MR_DEF_SUCC, NULL, 0);
				MrProKvFree(&pstKv, mysql_num_fields((MYSQL_RES*)pReturn));
			}
			else {
				ucReturn = MR_DEF_UNKN;
				goto MrSquidMysqlResult;
			}
		}
		else {
			ucReturn = MR_DEF_UNKN;
			goto MrSquidMysqlResult;
		}
	}
	else {
		goto MrSquidMysqlResult;
	}

	return;

MrSquidMysqlResult:
	if (NULL != (pstKv = (MRPROKV_S*)MrProKvMalloc(MR_PRO_1, MR_PACKET))) {
		if (MR_DEF_SUCC == ucReturn) {
			strcpy((char*)(pstKv + usField)->m_puszValue, "SUCCESS");
		}
		else {
			strcpy((char*)(pstKv + usField)->m_puszValue, "FAILURE");
		}
		(pstKv + usField)->m_ucKey = CPro.MrProGetVerb();
		(pstKv + usField)->m_iLength = strlen((char*)(pstKv + usField)->m_puszValue);
		usField += 1;
		MrSquidResultSend(CTcp, CPro, ucReturn, pstKv, usField);
		if (true == g_bDebug) {
			MR_DEBUG("KeyValue:\n");
			MrProKvDisplay(pstKv, MR_PRO_1);
		}
		MrProKvFree(&pstKv, MR_PRO_1);
	}

	return;
}

void MrSquidPgsqlCore(CMrSql* pCSql, CMrTcp& CTcp, CMrPro& CPro, char* pszSql)
{
	return;
}

void MrSquidOraclCore(CMrSql* pCSql, CMrTcp& CTcp, CMrPro& CPro, char* pszSql)
{
	return;
}

void MrSquidResultSend(CMrTcp& CTcp, CMrPro& rCPro, UCHAR ucReturn, MRPROKV_S* pstKv, USHORT usTotal)
{
	CMrPro CPro;
	int iLength = 0;
	UCHAR uszPacket[MR_ETHERNET - MR_MACIPTCP] = { 0 };

	CPro.MrProClone(rCPro);

	/* body */
	if (NULL != pstKv) {
		iLength = MrProKvOutput(pstKv, usTotal, &uszPacket[MR_PRO_HEAD], MR_ETHERNET - MR_MACIPTCP - MR_PRO_HEAD);
	}

	/* head */
	CPro.MrProPutResponse(MR_PRO_N);
	CPro.MrProPutExpand(MR_PRO_0);
	CPro.MrProPutResult(ucReturn);
	CPro.MrProPutLength(iLength);
	CPro.MrProPutTotal(usTotal);
	CPro.MrProPutBody(NULL);
	CPro.MrProCreate(uszPacket, MR_ETHERNET - MR_MACIPTCP);

	/* send */
	CTcp.MrNetSend(uszPacket, MR_PRO_HEAD + iLength);

	return;
}
